/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2020-2020. All rights reserved.
 */
/* -------------------------------------------------------------------------
 *
 * gs_imcs.cpp
 *	  openGauss define and remove imcs code.
 *
 * Portions Copyright (c) 2020 Huawei Technologies Co.,Ltd.
 * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 * Portions Copyright (c) 2010-2012 Postgres-XC Development Group
 * Portions Copyright (c) 2021, openGauss Contributors
 *
 *
 * IDENTIFICATION
 *	  src/gausskernel/optimizer/commands/gs_imcs.cpp
 *
 * -------------------------------------------------------------------------
 */
#include "c.h"
#include "postgres.h"
#include "knl/knl_variable.h"

#include <cstddef>
#include "access/cstore_delta.h"
#include "access/reloptions.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/index.h"
#include "catalog/indexing.h"
#include "catalog/pg_constraint.h"
#include "catalog/pg_opclass.h"
#include "catalog/pg_opfamily.h"
#include "catalog/pg_partition.h"
#include "catalog/pg_partition_fn.h"
#include "catalog/pg_tablespace.h"
#include "catalog/pg_type.h"
#include "catalog/gs_db_privilege.h"
#include "catalog/namespace.h"
#include "catalog/pg_namespace.h"
#include "commands/comment.h"
#include "commands/dbcommands.h"
#include "commands/defrem.h"
#include "catalog/heap.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "foreign/foreign.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
#include "optimizer/clauses.h"
#include "optimizer/planner.h"
#include "parser/parse_coerce.h"
#include "parser/parse_func.h"
#include "parser/parse_oper.h"
#ifdef PGXC
#include "optimizer/pgxcship.h"
#include "parser/parse_utilcmd.h"
#include "pgxc/pgxc.h"
#endif
#include <vector>
#include <algorithm>
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/tcap.h"
#include "tcop/utility.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/palloc.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "access/heapam.h"
#include "access/ustore/knl_whitebox_test.h"
#include "securec.h"
#include "catalog/gs_imcs.h"
#include "access/cstore_vector.h"
#include "utils/dynahash.h"
#include "postmaster/bgworker.h"
#include "funcapi.h"
#include "libpq/libpq-int.h"
#include "utils/guc_storage.h"
#ifdef ENABLE_HTAP
#include "storage/htap/htap_modify.h"
#include "access/imcs/imcs_insert.h"
#include "access/imcs/imcu.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "access/imcs/imcs_ctlg.h"
#endif
#include "pgxc/execRemote.h"
#include "libpq/libpq.h"

Datum gs_imcs_views(PG_FUNCTION_ARGS)
{
    Datum result;
#ifndef ENABLE_HTAP
    ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("htap disabled, not supported")));
    return result;
#else
    Form_gs_imcs gsimcs_form;
    IMCSScanContext *imcs_scan_ctx;
    errno_t rc;
    FuncCallContext *funcctx = NULL;
    Relation imcs_rel = NULL;
    TableScanDesc scan = NULL;
    HeapTuple heap_tup = NULL;
    IMCSHashEntry *result_found = NULL;
    HTAB *imcsHashTable = g_instance.stat_cxt.imcsHashTable;
    bool found = false;
    
    if (SRF_IS_FIRSTCALL()) {
        TupleDesc tupdesc = NULL;
        MemoryContext oldcontext = NULL;
        funcctx = SRF_FIRSTCALL_INIT();
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
        tupdesc = CreateTemplateTupleDesc(Natts_sys_view, false, TableAmHeap);

        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_sys_view_reloid, "reloid", OIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_sys_view_relname, "relname", NAMEOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_sys_view_imcsoid, "imcsoid", OIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_sys_view_imcsmetaname, "imcsmetaname", NAMEOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_sys_view_imcsparentid, "parentoid", OIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_sys_view_imcsispart, "imcsispart", BOOLOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_sys_view_imcsstatus, "imcsstatus", NAMEOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_sys_view_imcsnattr, "imcsnattr", INT2OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_sys_view_imcskey, "imcskey", INT2VECTOROID, -1, 0);

        imcs_rel = heap_open(ImcsRelationId, AccessShareLock);
        scan = tableam_scan_begin(imcs_rel, SnapshotNow, 0, NULL);
        /* imcs_scan_ctx is used to transit imcs_rel and scan through the SRF */
        imcs_scan_ctx = (IMCSScanContext*)palloc(sizeof(IMCSScanContext));
        imcs_scan_ctx->imcs_rel = imcs_rel;
        imcs_scan_ctx->scan = scan;
        funcctx->user_fctx = (void*)imcs_scan_ctx;
        funcctx->tuple_desc = BlessTupleDesc(tupdesc);
        (void)MemoryContextSwitchTo(oldcontext);
    }

    funcctx = SRF_PERCALL_SETUP();
    /* search gs_imcs table */
    imcs_scan_ctx = (IMCSScanContext*)funcctx->user_fctx;
    scan = imcs_scan_ctx->scan;
    imcs_rel = imcs_scan_ctx->imcs_rel;
    if ((heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection))!= NULL) {
        gsimcs_form = (Form_gs_imcs)GETSTRUCT(heap_tup);
        HeapTuple tuple = NULL;
        TupleDesc imcs_tuple_desc;
        Datum gs_imcs_values[Natts_sys_view]; 
        bool gs_imcs_nulls[Natts_sys_view] = {false};
        errno_t errorno = EOK;
        errorno = memset_s(gs_imcs_nulls, sizeof(gs_imcs_nulls), false, sizeof(gs_imcs_nulls));
        securec_check(errorno, "\0", "\0");
        /* search in imcsHashTable */
        result_found = (IMCSHashEntry *)hash_search(imcsHashTable, &(gsimcs_form->reloid), HASH_ENTER, &found);
        char *relname = (char*)palloc0(sizeof(char) * NAMEDATALEN);
        rc = snprintf_s(relname, NAMEDATALEN, NAMEDATALEN - 1, gsimcs_form->relname.data);
        securec_check_ss(rc, "", "");
        char *imcsmetaname = (char*)palloc0(sizeof(char) * NAMEDATALEN);
        rc = snprintf_s(imcsmetaname, NAMEDATALEN, NAMEDATALEN - 1, gsimcs_form->imcsmetaname.data);
        securec_check_ss(rc, "", "");
        imcs_tuple_desc = RelationGetDescr(imcs_rel);
        /* If can't find the record in imcsHashTable, then initialize the imcsstatus to INITIAL and add other
        information in gs_imcs to imcsHashTable */ 
        if (found == false) {
            AddToImcHashTbl(gsimcs_form->reloid, gsimcs_form->imcsoid, gsimcs_form->imcsispart, NULL, 0, InvalidXLogRecPtr);
            char *imcsstatus = (char*)palloc0(sizeof(char) * NAMEDATALEN);
            if (t_thrd.postmaster_cxt.HaShmData->current_mode == PRIMARY_MODE) {
                rc = snprintf_s(imcsstatus, NAMEDATALEN, NAMEDATALEN - 1, IMCS_STATUS_ONSTANDBY);
            } else {
                rc = snprintf_s(imcsstatus, NAMEDATALEN, NAMEDATALEN - 1, IMCS_STATUS_INITIAL);
            }
            securec_check_ss(rc, "", "");
            gs_imcs_values[Anum_sys_view_imcsstatus - 1] = NameGetDatum(imcsstatus);
        } else {
            char *imcsstatus = (char*)palloc0(sizeof(char) * NAMEDATALEN);
            rc = snprintf_s(imcsstatus, NAMEDATALEN, NAMEDATALEN - 1, result_found->imcsstatus);
            securec_check_ss(rc, "", "");
            gs_imcs_values[Anum_sys_view_imcsstatus - 1] = NameGetDatum(imcsstatus);
            result_found->imcsoid = gsimcs_form->imcsoid;
        }
        gs_imcs_values[Anum_sys_view_reloid - 1] = ObjectIdGetDatum(gsimcs_form->reloid);
        gs_imcs_values[Anum_sys_view_relname - 1] = NameGetDatum(relname);
        gs_imcs_values[Anum_sys_view_imcsoid - 1] = ObjectIdGetDatum(gsimcs_form->imcsoid);
        gs_imcs_values[Anum_sys_view_imcsmetaname - 1] = NameGetDatum(imcsmetaname);
        gs_imcs_values[Anum_sys_view_imcsparentid - 1] = ObjectIdGetDatum(gsimcs_form->parentoid);
        gs_imcs_values[Anum_sys_view_imcsispart - 1] = BoolGetDatum(gsimcs_form->imcsispart);
        gs_imcs_values[Anum_sys_view_imcsnattr - 1] = Int16GetDatum(gsimcs_form->imcsnattr);
        gs_imcs_values[Anum_sys_view_imcskey - 1] = PointerGetDatum(&gsimcs_form->imcskey);
        
        tuple = heap_form_tuple(funcctx->tuple_desc, gs_imcs_values, gs_imcs_nulls);
        result = HeapTupleGetDatum(tuple);
        SRF_RETURN_NEXT(funcctx, result);
    } else {
        tableam_scan_end(scan);
        heap_close(imcs_rel, NoLock);
        SRF_RETURN_DONE(funcctx);
    }
#endif
}

Datum gs_imcu_descs(PG_FUNCTION_ARGS)
{
    Datum result;
#ifndef ENABLE_HTAP
    ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("htap disabled, not supported")));
    return result;
#else
    FuncCallContext *funcctx = NULL;
    IMCSDescScanContext *imcs_desc_scan_ctx = NULL;
    IMCSDesc_Entry *imcs_table_entry = NULL;
    IMCUDesc_Entry *imcu_entry = NULL;
    
    /* stuff done only on the first call of the function */
    if (SRF_IS_FIRSTCALL()) {
        TupleDesc tupdesc = NULL;
        MemoryContext oldcontext = NULL;

        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();

        /*
         * switch to memory context appropriate for multiple function calls
         */
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

        LWLockAcquire(IMCU_CACHE->m_imcs_latch, LW_SHARED);

        /*
         * build tupdesc for result tuples. This must match this function's
         * pg_proc entry!
         */
        tupdesc = CreateTemplateTupleDesc(Natts_gs_imcu_descs, false, TableAmHeap);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_reloid, "reloid", OIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_xmin, "imu_xmin", XIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_id, "imcu_id", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_min, "imcu_min", TEXTOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_max, "imcu_max", TEXTOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_row_count, "row_count", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_size, "imcu_size", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_mode, "imcu_mode", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_pointer, "imcu_pointer", INT8OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_magic, "magic", INT4OID, -1, 0);

        funcctx->tuple_desc = BlessTupleDesc(tupdesc);

        imcs_desc_scan_ctx = (IMCSDescScanContext*)palloc(sizeof(IMCSDescScanContext));
        imcs_desc_scan_ctx->imcs_hash_seq = (HASH_SEQ_STATUS*)palloc0(sizeof(HASH_SEQ_STATUS));
        hash_seq_init(imcs_desc_scan_ctx->imcs_hash_seq, IMCU_CACHE->m_imcs_hash);
        imcs_table_entry = (IMCSDesc_Entry *)hash_seq_search(imcs_desc_scan_ctx->imcs_hash_seq);
        if (imcs_table_entry != NULL) {
            imcs_desc_scan_ctx->imcs_oid = imcs_table_entry->imcs_oid;
            imcs_desc_scan_ctx->imcu_hash_seq = (HASH_SEQ_STATUS *)palloc0(sizeof(HASH_SEQ_STATUS));
            hash_seq_init(imcs_desc_scan_ctx->imcu_hash_seq, imcs_table_entry->imcs_desc->im_cudesc_table);
        } else {
            LWLockRelease(IMCU_CACHE->m_imcs_latch);
            SRF_RETURN_DONE(funcctx);
        }
        funcctx->user_fctx = (void*)imcs_desc_scan_ctx;
        MemoryContextSwitchTo(oldcontext);
    }

    /* stuff done on every call of the function */
    funcctx = SRF_PERCALL_SETUP();

    imcs_desc_scan_ctx = (IMCSDescScanContext *)funcctx->user_fctx;
    imcu_entry = (IMCUDesc_Entry *)hash_seq_search(imcs_desc_scan_ctx->imcu_hash_seq);
        
    while (imcu_entry == NULL) {
        /* this imcs table was empty, try to scan another one */
        imcs_table_entry = (IMCSDesc_Entry *)hash_seq_search(imcs_desc_scan_ctx->imcs_hash_seq);
        /* all imcs_desc_table was empty, finish here */
        if (imcs_table_entry == NULL) {
            LWLockRelease(IMCU_CACHE->m_imcs_latch);
            SRF_RETURN_DONE(funcctx);
        }
        imcs_desc_scan_ctx->imcs_oid = imcs_table_entry->imcs_oid;
        hash_seq_init(imcs_desc_scan_ctx->imcu_hash_seq, imcs_table_entry->imcs_desc->im_cudesc_table);
        imcu_entry = (IMCUDesc_Entry *)hash_seq_search(imcs_desc_scan_ctx->imcu_hash_seq);
        if (imcu_entry != NULL) {
            break;
        }
    }

    HeapTuple tuple = NULL;
    Datum gs_imcu_values[Natts_gs_imcu_descs]; 
    bool gs_imcu_nulls[Natts_gs_imcu_descs];
    errno_t errorno = EOK;
    errorno = memset_s(gs_imcu_nulls, sizeof(gs_imcu_nulls), false, sizeof(gs_imcu_nulls));
    securec_check(errorno, "\0", "\0");
    
    gs_imcu_values[Anum_gs_imcu_reloid - 1] = ObjectIdGetDatum(imcs_desc_scan_ctx->imcs_oid);
    gs_imcu_values[Anum_gs_imcu_xmin - 1] = TransactionIdGetDatum(imcu_entry->cu_desc->xmin);
    gs_imcu_values[Anum_gs_imcu_id - 1] = Int32GetDatum(imcu_entry->cu_desc->cu_id);
    gs_imcu_values[Anum_gs_imcu_min - 1] = CStringGetTextDatum(imcu_entry->cu_desc->cu_min);
    gs_imcu_values[Anum_gs_imcu_max - 1] = CStringGetTextDatum(imcu_entry->cu_desc->cu_max);
    gs_imcu_values[Anum_gs_imcu_row_count - 1] = Int32GetDatum(imcu_entry->cu_desc->row_count);
    gs_imcu_values[Anum_gs_imcu_size - 1] = Int32GetDatum(imcu_entry->cu_desc->cu_size);
    gs_imcu_values[Anum_gs_imcu_mode - 1] = Int32GetDatum(imcu_entry->cu_desc->cu_mode);
    gs_imcu_values[Anum_gs_imcu_pointer - 1] = Int64GetDatum(imcu_entry->cu_desc->cu_pointer);
    gs_imcu_values[Anum_gs_imcu_magic - 1] = Int32GetDatum(imcu_entry->cu_desc->magic);
    tuple = heap_form_tuple(funcctx->tuple_desc, gs_imcu_values, gs_imcu_nulls);
    result = HeapTupleGetDatum(tuple);
    SRF_RETURN_NEXT(funcctx, result);
#endif
}

Datum gs_imcs_mem_status(PG_FUNCTION_ARGS)
{
    Datum result;
#ifndef ENABLE_HTAP
    ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("htap disabled, not supported")));
    return result;
#else
    FuncCallContext *funcctx = NULL;
    IMCSDescScanContext *imcs_desc_scan_ctx = NULL;
    IMCSDesc_Entry *imcs_table_entry = NULL;

    /* stuff done only on the first call of the function */
    if (SRF_IS_FIRSTCALL()) {
        TupleDesc tupdesc = NULL;
        MemoryContext oldcontext = NULL;

        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();

        /*
         * switch to memory context appropriate for multiple function calls
         */
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

        LWLockAcquire(IMCU_CACHE->m_imcs_latch, LW_SHARED);

        /*
         * build tupdesc for result tuples. This must match this function's
         * pg_proc entry!
         */
        tupdesc = CreateTemplateTupleDesc(Natts_gs_imcs_mem_status, false, TableAmHeap);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_imcs_desc_reloid, "reloid", OIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcs_desc_rowgroup_num, "rowgroup_num", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcs_desc_mem_num, "cu_num_in_mem", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcs_desc_mem_size, "cu_mem_size", INT8OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcs_desc_disk_num, "cu_num_in_disk", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcs_desc_disk_size, "cu_disk_size", INT8OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcs_desc_last_flushed_csn, "last_flush_csn", INT8OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcs_desc_delta_row_count, "delta_row_count", INT8OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcs_desc_delta_mem_size, "delta_mem_size", INT8OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcs_delta_visible_rows, "delta_visible_rows", INT8OID, -1, 0);

        funcctx->tuple_desc = BlessTupleDesc(tupdesc);

        imcs_desc_scan_ctx = (IMCSDescScanContext*)palloc(sizeof(IMCSDescScanContext));
        imcs_desc_scan_ctx->imcs_hash_seq = (HASH_SEQ_STATUS*)palloc0(sizeof(HASH_SEQ_STATUS));
        hash_seq_init(imcs_desc_scan_ctx->imcs_hash_seq, IMCU_CACHE->m_imcs_hash);
        funcctx->user_fctx = (void*)imcs_desc_scan_ctx;
        MemoryContextSwitchTo(oldcontext);
    }

    /* stuff done on every call of the function */
    funcctx = SRF_PERCALL_SETUP();

    imcs_desc_scan_ctx = (IMCSDescScanContext *)funcctx->user_fctx;

    imcs_table_entry = (IMCSDesc_Entry *)hash_seq_search(imcs_desc_scan_ctx->imcs_hash_seq);
    if (imcs_table_entry != NULL) {
        HeapTuple tuple = NULL;
        Datum gs_imcs_values[Natts_gs_imcs_mem_status];
        bool gs_imcs_nulls[Natts_gs_imcs_mem_status];
        errno_t errorno = EOK;
        errorno = memset_s(gs_imcs_nulls, sizeof(gs_imcs_nulls), false, sizeof(gs_imcs_nulls));
        securec_check(errorno, "\0", "\0");
        Oid relid = imcs_table_entry->imcs_desc->imcs_oid;
        uint64_t delta_mem_size = 0;
        uint64_t delta_row_count = 0;
        uint64_t delta_visible_rows = 0;

        HTAPSearchDelta(relid, delta_row_count, delta_visible_rows, delta_mem_size);
        gs_imcs_values[Anum_imcs_desc_reloid - 1] = ObjectIdGetDatum(relid);
        gs_imcs_values[Anum_gs_imcs_desc_rowgroup_num - 1] = Int32GetDatum(imcs_table_entry->imcs_desc->rowgroup_num);
        gs_imcs_values[Anum_gs_imcs_desc_mem_num - 1] = Int32GetDatum(imcs_table_entry->imcs_desc->cu_num_in_mem);
        gs_imcs_values[Anum_gs_imcs_desc_mem_size - 1] = Int64GetDatum(imcs_table_entry->imcs_desc->cu_mem_size);
        gs_imcs_values[Anum_gs_imcs_desc_disk_num - 1] = Int32GetDatum(imcs_table_entry->imcs_desc->cu_num_in_disk);
        gs_imcs_values[Anum_gs_imcs_desc_disk_size - 1] = Int64GetDatum(imcs_table_entry->imcs_desc->cu_disk_size);
        gs_imcs_values[Anum_gs_imcs_desc_last_flushed_csn - 1] = Int64GetDatum(imcs_table_entry->imcs_desc->last_flushed_csn);
        gs_imcs_values[Anum_gs_imcs_desc_delta_row_count - 1] = Int64GetDatum(delta_row_count);
        gs_imcs_values[Anum_gs_imcs_desc_delta_mem_size - 1] = Int64GetDatum(delta_mem_size);
        gs_imcs_values[Anum_gs_imcs_delta_visible_rows - 1] = Int64GetDatum(delta_visible_rows);
        tuple = heap_form_tuple(funcctx->tuple_desc, gs_imcs_values, gs_imcs_nulls);
        result = HeapTupleGetDatum(tuple);
        SRF_RETURN_NEXT(funcctx, result);
    } else {
        LWLockRelease(IMCU_CACHE->m_imcs_latch);
        SRF_RETURN_DONE(funcctx);
    }
#endif
}

Datum gs_imcu_slot_status(PG_FUNCTION_ARGS)
{
    Datum result;
#ifndef ENABLE_HTAP
    ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("htap disabled, not supported")));
    return result;
#else
    FuncCallContext *funcctx = NULL;
    IMCSCacheScanContext *imcs_cache_scan_ctx = NULL;

    /* stuff done only on the first call of the function */
    if (SRF_IS_FIRSTCALL()) {
        TupleDesc tupdesc = NULL;
        MemoryContext oldcontext = NULL;

        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();

        /*
         * switch to memory context appropriate for multiple function calls
         */
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

        LWLockAcquire(IMCU_CACHE->m_imcs_latch, LW_SHARED);

        /*
         * build tupdesc for result tuples. This must match this function's
         * pg_proc entry!
         */
        tupdesc = CreateTemplateTupleDesc(Natts_gs_imcu_slot_status, false, TableAmHeap);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_imcu_slot_reloid, "reloid", OIDOID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_slot_cu_id, "cu_id", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_slot_col_id, "col_id", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_slot_id, "slot_id", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_slot_datablock_size, "block_size", INT4OID, -1, 0);
        TupleDescInitEntry(tupdesc, (AttrNumber)Anum_gs_imcu_slot_flag, "flag", CHAROID, -1, 0);

        funcctx->tuple_desc = BlessTupleDesc(tupdesc);

        imcs_cache_scan_ctx = (IMCSCacheScanContext*)palloc(sizeof(IMCSCacheScanContext));
        imcs_cache_scan_ctx->cur_slot_id = 0;
        imcs_cache_scan_ctx->max_slot_id = IMCU_CACHE->GetCacheSlotNum();
        funcctx->user_fctx = (void*)imcs_cache_scan_ctx;
        MemoryContextSwitchTo(oldcontext);
    }

    /* stuff done on every call of the function */
    funcctx = SRF_PERCALL_SETUP();
    imcs_cache_scan_ctx = (IMCSCacheScanContext *)funcctx->user_fctx;

    while (imcs_cache_scan_ctx->cur_slot_id <= imcs_cache_scan_ctx->max_slot_id) {
        HeapTuple tuple = NULL;
        Datum gs_imcs_cache_values[Natts_gs_imcu_slot_status];
        bool gs_imcs_cache_nulls[Natts_gs_imcu_slot_status];
        errno_t errorno = EOK;
        errorno = memset_s(gs_imcs_cache_nulls, sizeof(gs_imcs_cache_nulls), false, sizeof(gs_imcs_cache_nulls));
        securec_check(errorno, "\0", "\0");

        CacheDesc *desc = IMCU_CACHE->GetCacheSlotDesc(imcs_cache_scan_ctx->cur_slot_id);
        imcs_cache_scan_ctx->cur_slot_id++;

        if (desc->m_flag == CACHE_COlUMN_DATA) {
            DataSlotTag data_slot_tag;
            errno_t rc = memcpy_s(&data_slot_tag.slotTag, MAX_CACHE_TAG_LEN, desc->m_cache_tag.key, sizeof(DataSlotTagKey));
            securec_check(rc, "\0", "\0");
            gs_imcs_cache_values[Anum_imcu_slot_reloid - 1] = ObjectIdGetDatum(data_slot_tag.slotTag.cuSlotTag.m_rnode.relNode);
            gs_imcs_cache_values[Anum_gs_imcu_slot_cu_id - 1] = Int32GetDatum(data_slot_tag.slotTag.cuSlotTag.m_CUId);
            gs_imcs_cache_values[Anum_gs_imcu_slot_col_id - 1] = Int32GetDatum(data_slot_tag.slotTag.cuSlotTag.m_colId);
            gs_imcs_cache_values[Anum_gs_imcu_slot_id - 1] = Int32GetDatum(desc->m_slot_id);
            gs_imcs_cache_values[Anum_gs_imcu_slot_datablock_size - 1] = Int32GetDatum(desc->m_datablock_size);
            gs_imcs_cache_values[Anum_gs_imcu_slot_flag - 1] = CharGetDatum(desc->m_flag);

            tuple = heap_form_tuple(funcctx->tuple_desc, gs_imcs_cache_values, gs_imcs_cache_nulls);
            result = HeapTupleGetDatum(tuple);
            SRF_RETURN_NEXT(funcctx, result);
        }
    }
    LWLockRelease(IMCU_CACHE->m_imcs_latch);
    SRF_RETURN_DONE(funcctx);
#endif
}

#ifdef ENABLE_HTAP
/* insert mapping relationship between rel and imcs table into system table gs_imcs table */
void gs_imcs_insert_tuple(Oid rel_oid, char* rel_name, Oid imcs_oid, char* imcs_meta_name,
                                  bool is_partition, Oid parentid, int2 nattr, int2vector* imcskey)
{
    errno_t rc;
    Relation gs_imcs_rel;
    HeapTuple gs_imcs_heap_tup;
    bool gs_imcs_nulls[Natts_gs_imcs];
    Datum gs_imcs_values[Natts_gs_imcs];
    gs_imcs_rel = heap_open(ImcsRelationId, RowExclusiveLock);

    rc = memset_s(gs_imcs_values, sizeof(gs_imcs_values), 0, sizeof(gs_imcs_values));
    securec_check(rc, "\0", "\0");
    rc = memset_s(gs_imcs_nulls, sizeof(gs_imcs_nulls), false, sizeof(gs_imcs_nulls));
    securec_check(rc, "\0", "\0");

    char* relname = (char*)palloc0(sizeof(char) * NAMEDATALEN);
    rc = snprintf_s(relname, NAMEDATALEN, NAMEDATALEN - 1, rel_name);
    securec_check_ss(rc, "", "");

    gs_imcs_values[Anum_gs_imcs_reloid - 1] = ObjectIdGetDatum(rel_oid);
    gs_imcs_values[Anum_gs_imcs_relname - 1] = NameGetDatum(relname);
    gs_imcs_values[Anum_gs_imcs_imcsoid - 1] = ObjectIdGetDatum(imcs_oid);
    gs_imcs_values[Anum_gs_imcs_imcsmetaname - 1] = NameGetDatum(imcs_meta_name);
    gs_imcs_values[Anum_gs_imcs_imcsispart - 1] = BoolGetDatum(is_partition);
    gs_imcs_values[Anum_gs_imcs_imcsparentid - 1] = ObjectIdGetDatum(parentid);
    gs_imcs_values[Anum_gs_imcs_imcsnattr - 1] = ObjectIdGetDatum(nattr);
    gs_imcs_values[Anum_gs_imcs_imcskey - 1] = PointerGetDatum(imcskey);
    gs_imcs_heap_tup = heap_form_tuple(gs_imcs_rel->rd_att, gs_imcs_values, gs_imcs_nulls);

    /* Do the insertion */
    (void)simple_heap_insert(gs_imcs_rel, gs_imcs_heap_tup);
    CatalogUpdateIndexes(gs_imcs_rel, gs_imcs_heap_tup);

    /* Make the changes visible */
    CommandCounterIncrement();

    heap_freetuple_ext(gs_imcs_heap_tup);
    heap_close(gs_imcs_rel, NoLock);
    return;
}

/* delete mapping relationship between rel and imcs table into system table gs_imcs table */
void gs_imcs_delete_tuple(Oid imcs_oid, std::unordered_map<Oid, RelFileNode>* oid_del_info)
{
    Oid found_Oid = InvalidOid;
    Relation relation = NULL;
    TableScanDesc scan;
    Form_gs_imcs gsimcs_form;
    HeapTuple heap_tup = NULL;
    RelFileNode relInf;

    memset(&relInf, 0, sizeof(RelFileNode));
    relInf.bucketNode = -1;

    /* search imcs_oid in gs_imcs table */
    relation = heap_open(ImcsRelationId, RowExclusiveLock);
    scan = tableam_scan_begin(relation, SnapshotNow, 0, NULL);
    heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
    while (heap_tup != NULL) {
        gsimcs_form = (Form_gs_imcs)GETSTRUCT(heap_tup);
        if (gsimcs_form->imcsoid == imcs_oid) {
            /* found */
            found_Oid = gsimcs_form->imcsoid;
            break;
        }
        heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
    }

    if (HeapTupleIsValid(heap_tup)) {
        simple_heap_delete(relation, &heap_tup->t_self);
        if (oid_del_info) {
            if ((*oid_del_info).find(gsimcs_form->reloid) == (*oid_del_info).end()) {
                (*oid_del_info)[gsimcs_form->reloid] = relInf;
            }
        } else {
            RemoveFromImcHashTbl(gsimcs_form->reloid);
        }

        /* clear up correlative imcs in pg_class catalog */
        if (HeapTupleIsValid(ScanPgRelation(found_Oid, true, false))) {
            ObjectAddress imcs_object;
            imcs_object.classId = RelationRelationId;
            imcs_object.objectId = found_Oid;
            imcs_object.objectSubId = 0;
            performDeletion(&imcs_object, DROP_RESTRICT, PERFORM_DELETION_INTERNAL);
        } else {
            ereport(DEBUG2,
                    (errmodule(MOD_OPT), errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                            errmsg("imcs oid %d relation is invalid when delete it.", (int)found_Oid)));
        }
    }

    tableam_scan_end(scan);
    heap_close(relation, NoLock);

    /* Make the changes visible */
    if(!oid_del_info) {
        CommandCounterIncrement();
    }

    return;
}

Oid search_table_from_gs_imcs(Oid rel_oid)
{
    Oid found_Oid = InvalidOid;

    if (t_thrd.postmaster_cxt.HaShmData->current_mode == STANDBY_MODE) {
        IMCSHashEntry *result_found = NULL;
        bool found = false;
        result_found =
                (IMCSHashEntry *)hash_search(g_instance.stat_cxt.imcsHashTable, &(rel_oid), HASH_FIND, &found);
        if (found) {
            return result_found->rel_oid;
        } else {
            return InvalidOid;
        }
    }

    Relation relation = NULL;
    TableScanDesc scan;
    Form_gs_imcs gsimcs_form;
    HeapTuple heap_tup = NULL;

    /* search imcs_oid in gs_imcs table */
    relation = heap_open(ImcsRelationId, AccessShareLock);
    scan = tableam_scan_begin(relation, SnapshotNow, 0, NULL);
    heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
    while (heap_tup != NULL) {
        gsimcs_form = (Form_gs_imcs)GETSTRUCT(heap_tup);
        if (gsimcs_form->reloid == rel_oid) {
            /* found */
            found_Oid = gsimcs_form->imcsoid;
            break;
        }
        heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
    }
    tableam_scan_end(scan);
    heap_close(relation, NoLock);
    return found_Oid;
}

Form_gs_imcs search_records_from_gs_imcs(Oid rel_oid)
{
    Relation relation = NULL;
    TableScanDesc scan;
    Form_gs_imcs gsimcs_form = NULL;
    HeapTuple heap_tup = NULL;

    /* search record in gs_imcs table */
    relation = heap_open(ImcsRelationId, AccessShareLock);
    scan = tableam_scan_begin(relation, SnapshotNow, 0, NULL);
    heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
    while (heap_tup != NULL) {
        gsimcs_form = (Form_gs_imcs)GETSTRUCT(heap_tup);
        if (gsimcs_form->reloid == rel_oid) {
            /* found */
            break;
        }
        heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
    }
    if (!gsimcs_form || gsimcs_form->reloid != rel_oid) {
        tableam_scan_end(scan);
        heap_close(relation, NoLock);
        return NULL;
    }

    tableam_scan_end(scan);
    heap_close(relation, NoLock);
    return gsimcs_form;
}

Oid search_rel_oid_from_hashtable(Oid rel_oid)
{
    Oid found_oid = InvalidOid;
    IMCSHashEntry* result_found = NULL;
    bool found = false;
    //whether relation is stored in IMCS 
    result_found = (IMCSHashEntry*)hash_search(g_instance.stat_cxt.imcsHashTable,
                                               &(rel_oid), HASH_FIND, &found);
    if (found != false)
    {
        found_oid = result_found -> rel_oid;
    }

    return found_oid;
}

bool search_parent_oid_from_gs_imcs_for_partition(Oid parent_oid, Oid* imcs_oid)
{
    Oid found_Oid = InvalidOid;
    Relation relation = NULL;
    TableScanDesc scan;
    Form_gs_imcs gsimcs_form;
    HeapTuple heap_tup = NULL;
    bool is_found = false;
    int tup_counter = 0;

    relation = heap_open(ImcsRelationId, AccessShareLock);
    scan = tableam_scan_begin(relation, SnapshotNow, 0, NULL);
    heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
    while (heap_tup != NULL) {
        gsimcs_form = (Form_gs_imcs)GETSTRUCT(heap_tup);
        if (gsimcs_form->parentoid == parent_oid) {
            /* found */
            tup_counter++;
            found_Oid = gsimcs_form->imcsoid;
        }
        heap_tup = (HeapTuple)tableam_scan_getnexttuple(scan, ForwardScanDirection);
    }

    if ((tup_counter == 1) && (gsimcs_form->reloid == parent_oid)) {
        is_found = true;
        *imcs_oid = found_Oid;
    }
    tableam_scan_end(scan);
    heap_close(relation, NoLock);
    return is_found;
}
/* delete heap tuple from pg_class, pg_attribute and gs_imcs for imcs */
void delete_gs_imcs_table_on_rel(Relation rel, std::unordered_map<Oid, RelFileNode>* oid_del_info)
{
    Oid imcs_oid = InvalidOid;

    /* partitions and subpartitions */
    if (RELATION_IS_PARTITIONED(rel)) {
        List* partition_oid_list = NIL;
        ListCell* cell = NULL;
        partition_oid_list = relationGetPartitionOidList(rel);  //partition oid list
        foreach(cell, partition_oid_list) {
            Oid part_oid = lfirst_oid(cell);
            /* search imcs_oid */
            imcs_oid = search_table_from_gs_imcs(part_oid);
            if (OidIsValid(imcs_oid)) {
                /* remove mapping info from gs_imcs and pg_class tables */
                gs_imcs_delete_tuple(imcs_oid, oid_del_info);
            }

            List* subpart_oid_list = PartOidGetSubPartitionOidList(rel, part_oid);  //subpartition oid list
            ListCell* subcell = NULL;
            foreach(subcell, subpart_oid_list) {
                Oid subpart_oid = lfirst_oid(subcell);
                /* search imcs_oid */
                imcs_oid = search_table_from_gs_imcs(subpart_oid);
                if (OidIsValid(imcs_oid)) {
                    /* remove mapping info from gs_imcs and pg_class tables */
                    gs_imcs_delete_tuple(imcs_oid, oid_del_info);
                }
            }
            if (subpart_oid_list != NULL) {
                releasePartitionOidList(&subpart_oid_list);
            }
        }
        if (partition_oid_list != NULL) {
            releasePartitionOidList(&partition_oid_list);
        }
    }
    /* rel imcs oid */
    imcs_oid = search_table_from_gs_imcs(RelationGetRelid(rel));
    /* remove mapping info from gs_imcs and pg_class tables */
    if (OidIsValid(imcs_oid)) {
        gs_imcs_delete_tuple(imcs_oid, oid_del_info);
    }
    return;
}
/* IMCS */
#endif